package com.ksyun.campus.metaserver.mq.consumer;


import com.alibaba.fastjson2.JSON;
import com.ksyun.campus.metaserver.common.result.BaseResult;
import com.ksyun.campus.metaserver.entity.dto.req.WriteReqDto;
import com.ksyun.campus.metaserver.entity.meta.StatInfo;
import com.ksyun.campus.metaserver.mq.event.SyncMetaDataEvent;
import com.ksyun.campus.metaserver.service.SyncMetaService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import static com.ksyun.campus.metaserver.common.constant.MqConstant.*;

/**
 * 从节点同步元数据消息消费者
 */

@Slf4j
@Component(value = "syncMetaDataConsumer")
@RequiredArgsConstructor
@RocketMQMessageListener(
        topic = SYNC_META_DATA_TOPIC,
        consumerGroup = SYNC_META_DATA_CONSUMER_GROUP,
        consumeThreadMax = 1
)
public class SyncMetaDataConsumer implements RocketMQListener<SyncMetaDataEvent> {

    private final SyncMetaService syncMetaService;

    @Override
    public void onMessage(SyncMetaDataEvent syncMetaDataEvent) {
        log.info("进入消费者");
        String requestOption = syncMetaDataEvent.getRequestOption();
        String fileSystemName = syncMetaDataEvent.getFileSystemName();
        String path = syncMetaDataEvent.getPath();
        StatInfo statInfo = syncMetaDataEvent.getStatInfo();
        switch (requestOption) {
            case CREATE:
                log.info("开始同步元数据create请求");
                BaseResult<StatInfo> createResp;
                try {
                    createResp = syncMetaService.createFile(fileSystemName, path, statInfo);
                } catch (Exception e) {
                    throw new RuntimeException(String.format("同步元数据create失败 [syncMetaDataEvent]: %s [ex]: %s", JSON.toJSONString(syncMetaDataEvent), e.getMessage()));
                }
                if (createResp.getCode() == 200) {
                    log.info("同步元数据create成功");
                } else {
                    throw new RuntimeException(String.format("同步元数据create失败 [syncMetaDataEvent]: %s [ex]: %s", JSON.toJSONString(syncMetaDataEvent), createResp.getMsg()));
                }
                break;
            case WRITE:
                log.info("开始同步元数据write请求");
                WriteReqDto writeReqDto = WriteReqDto.builder()
                        .statInfo(statInfo)
                        .build();
                BaseResult<String> writeResp;
                try {
                    writeResp = syncMetaService.commitWrite(fileSystemName, writeReqDto);
                } catch (Exception e) {
                    throw new RuntimeException(String.format("同步元数据write失败 [syncMetaDataEvent]: %s [ex]: %s", JSON.toJSONString(syncMetaDataEvent), e.getMessage()));
                }
                if (writeResp.getCode() == 200) {
                    log.info("同步元数据write成功");
                } else {
                    throw new RuntimeException(String.format("同步元数据write失败 [syncMetaDataEvent]: %s [ex]: %s", JSON.toJSONString(syncMetaDataEvent), writeResp.getMsg()));
                }
                break;
            case MKDIR:
                log.info("开始同步元数据mkdir请求");
                BaseResult<StatInfo> mkdirResp;
                try {
                    mkdirResp = syncMetaService.mkdir(fileSystemName, path, statInfo);
                } catch (Exception e) {
                    throw new RuntimeException(String.format("同步元数据mkdir失败 [syncMetaDataEvent]: %s [ex]: %s", JSON.toJSONString(syncMetaDataEvent), e.getMessage()));
                }
                if (mkdirResp.getCode() == 200) {
                    log.info("同步元数据mkdir成功");
                } else {
                    throw new RuntimeException(String.format("同步元数据mkdir失败 [syncMetaDataEvent]: %s [ex]: %s", JSON.toJSONString(syncMetaDataEvent), mkdirResp.getMsg()));
                }
                break;
            case DELETE:
                log.info("开始同步元数据delete请求");
                BaseResult<StatInfo> deleteResp;
                try {
                    deleteResp = syncMetaService.delete(fileSystemName, path);
                } catch (Exception e) {
                    throw new RuntimeException(String.format("同步元数据delete失败 [syncMetaDataEvent]: %s [ex]: %s", JSON.toJSONString(syncMetaDataEvent), e.getMessage()));
                }
                if (deleteResp.getCode() == 200) {
                    log.info("同步元数据delete成功");
                } else {
                    throw new RuntimeException(String.format("同步元数据delete失败 [syncMetaDataEvent]: %s [ex]: %s", JSON.toJSONString(syncMetaDataEvent), deleteResp.getMsg()));
                }
                break;
            case DELETE_ERROR:
                log.info("开始同步元数据deleteError请求");
                BaseResult<String> deleteErrorResp;
                try {
                    deleteErrorResp = syncMetaService.deleteErrorRecovery(fileSystemName, statInfo);
                } catch (Exception e) {
                    throw new RuntimeException(String.format("同步元数据deleteError失败 [syncMetaDataEvent]: %s [ex]: %s", JSON.toJSONString(syncMetaDataEvent), e.getMessage()));
                }
                if (deleteErrorResp.getCode() == 200) {
                    log.info("同步元数据deleteError成功");
                } else {
                    throw new RuntimeException(String.format("同步元数据deleteError失败 [syncMetaDataEvent]: %s [ex]: %s", JSON.toJSONString(syncMetaDataEvent), deleteErrorResp.getMsg()));
                }
                break;
            default:
                throw new RuntimeException(String.format("请求的操作无法识别 [requestOption]: %s [syncMetaDataEvent]: %s", requestOption, JSON.toJSONString(syncMetaDataEvent)));
        }

    }
}
